/*
 * @Version: 0.0.1
 * @Author: ider
 * @Date: 2021-05-14 19:09:43
 * @LastEditors: ider
 * @LastEditTime: 2021-05-22 01:12:34
 * @Description:
 */
package mongodb

import (
	"context"
	"fmt"
	"historyOfWikipedia/internal/model"
	"sort"
	"sync"
	"time"

	log "github.com/sirupsen/logrus"

	"go.mongodb.org/mongo-driver/bson"
	"go.mongodb.org/mongo-driver/bson/primitive"
	"go.mongodb.org/mongo-driver/mongo"
	"go.mongodb.org/mongo-driver/mongo/options"
)

type MongoDataBase struct {
	ctx                                  context.Context
	client                               *mongo.Client
	collection_pages_articles            *mongo.Collection
	collection_revision_detail           *mongo.Collection
	collection_revision_info             *mongo.Collection
	collection_revision_info_schedule    *mongo.Collection
	collection_snapshot_article_revision *mongo.Collection
	collection_revision_detail_schedule  *mongo.Collection
}

func failOnError(err error, msg string) {
	if err != nil {
		log.Fatalf("%s: %s", msg, err)
	}
}

func NewMongoClient(MongoUri string) *MongoDataBase {
	client, err := mongo.NewClient(options.Client().ApplyURI(MongoUri))
	if err != nil {
		log.Fatal(err)
	}
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	client.Connect(ctx)
	err = client.Ping(ctx, nil)
	if err != nil {
		log.Fatal(err)
	}
	database := client.Database("wikipedia_history")
	mongodb := MongoDataBase{
		ctx:                                  context.Background(),
		client:                               client,
		collection_pages_articles:            database.Collection("pages_articles"),
		collection_revision_detail:           database.Collection("revision_detail"),
		collection_revision_info:             database.Collection("revision_info"),
		collection_revision_info_schedule:    database.Collection("schedule_revision_info"),
		collection_snapshot_article_revision: database.Collection("snapshot_article_revision"),
		collection_revision_detail_schedule:  database.Collection("schedule_revision_detail"),
	}
	mongodb.initIndex()
	return &mongodb
}

func (c *MongoDataBase) initIndex() {
	mods := []mongo.IndexModel{
		{Keys: bson.M{"title": 1}},
		{Keys: bson.M{"redirect": 1}},
	}
	c.collection_pages_articles.Indexes().CreateMany(c.ctx, mods)
	mods = []mongo.IndexModel{
		{Keys: bson.M{"revid": 1}},
		{Keys: bson.M{"articleID": 1}},
		{Keys: bson.M{"timestamp": 1}},
	}
	c.collection_revision_detail.Indexes().CreateMany(c.ctx, mods)
	c.collection_revision_info.Indexes().CreateMany(c.ctx, mods)

	mods = []mongo.IndexModel{
		{Keys: bson.M{"update": 1}},
		{Keys: bson.M{"firstRevisionUpdate": 1}},
		{Keys: bson.M{"lastRevisionUpdate": 1}},
	}
	c.collection_revision_info_schedule.Indexes().CreateMany(c.ctx, mods)

	mods = []mongo.IndexModel{
		{Keys: bson.M{"lastSnapshot": 1}},
		{Keys: bson.M{"birthday": 1}},
		{Keys: bson.M{"lastRevisionUpdate": 1}},
	}
	c.collection_snapshot_article_revision.Indexes().CreateMany(c.ctx, mods)

	mods = []mongo.IndexModel{
		{Keys: bson.D{bson.E{Key: "articleID", Value: 1}, bson.E{Key: "revid", Value: 1}}},
		{Keys: bson.M{"articleID": 1}},
		{Keys: bson.M{"revid": 1}},
	}
	c.collection_revision_detail_schedule.Indexes().CreateMany(c.ctx, mods)
}

// Update_Pages_Articles 成功更新就返回 true
func (c *MongoDataBase) Update_Pages_Articles(pao *model.PagesArticlesObj) (status bool) {

	retOne := c.collection_pages_articles.FindOne(c.ctx, bson.M{"_id": pao.ID})
	// 没有文档，直接插入新的内容
	if retOne.Err() == mongo.ErrNoDocuments {
		_, err := c.collection_pages_articles.InsertOne(c.ctx, pao)
		if err == nil {
			status = true
		} else {
			log.Warn("insert fail", err)
			status = false
		}

	} else {
		var docInDB model.PagesArticlesObj
		retOne.Decode(&docInDB)
		if docInDB.Version > pao.Version {
			status = false
		} else {
			bsond := bson.D{bson.E{Key: "version", Value: pao.Version}}
			bson_unset := bson.D{}
			if pao.Ns != docInDB.Ns {
				bsond = append(bsond, bson.E{Key: "ns", Value: pao.Ns})
			}
			if pao.Redirect != docInDB.Redirect {
				// 清除已经消失的 UNdirect
				if pao.Redirect == "" {
					bson_unset = append(bson_unset, bson.E{Key: "redirect", Value: 1})
				} else {
					bsond = append(bsond, bson.E{Key: "redirect", Value: pao.Redirect})
				}
			}
			if pao.RevisionID != docInDB.RevisionID {
				bsond = append(bsond, bson.E{Key: "revisionID", Value: pao.RevisionID})
			}
			if pao.Title != docInDB.Title {
				log.Warn(fmt.Sprintf("title 发生改变:oid:%d,oname:%s,nname:%s", docInDB.ID, docInDB.Title, pao.Title))
				bsond = append(bsond, bson.E{Key: "title", Value: pao.Title})
			}
			var mongoUpdate primitive.M
			if len(bson_unset) == 0 {
				mongoUpdate = bson.M{"$set": bsond}
			} else {
				mongoUpdate = bson.M{"$set": bsond, "$unset": bson_unset}
			}
			_, err := c.collection_pages_articles.UpdateByID(c.ctx, pao.ID, mongoUpdate)
			if err != nil {
				log.Warn(err)
			}
			status = true
		}
	}
	return
}

// Update_Revision_Info 成功更新就返回 true
func (c *MongoDataBase) Update_Revision_Info(rio *model.RevisionInfoObj) (status bool, revisionObjectID primitive.ObjectID, err error) {
	// 记录不可更改，如果已经存在就不再更新
	showIDoption := options.FindOne().SetProjection(bson.M{"_id": 1})
	retOne := c.collection_revision_info.FindOne(c.ctx, bson.M{"revid": rio.Revid, "articleID": rio.ArticleID}, showIDoption)
	// 没有文档，直接插入新的内容
	if retOne.Err() == mongo.ErrNoDocuments {
		InsertOneResult, err := c.collection_revision_info.InsertOne(c.ctx, rio)
		if err == nil {
			status = true
			revisionObjectID = InsertOneResult.InsertedID.(primitive.ObjectID)
		} else {
			log.Warn("insert fail", err)
			status = false
		}
	} else {
		status = false
		var findoneobj model.RevisionInfoObj
		retOne.Decode(&findoneobj)
		revisionObjectID = findoneobj.ID
	}
	return
}

// Update_Revision_Info 成功更新就返回 true
func (c *MongoDataBase) Update_Revision_Detail(rdo *model.RevisionDetailObj, revisionObjectID primitive.ObjectID) (status bool, err error) {
	// 检查 detail
	showIDoption := options.FindOne().SetProjection(bson.M{"_id": 1})
	retOne := c.collection_revision_detail.FindOne(c.ctx, bson.M{"_id": revisionObjectID}, showIDoption)
	// 没有文档，直接插入新的内容
	if retOne.Err() == mongo.ErrNoDocuments {
		_, err := c.collection_revision_detail.InsertOne(c.ctx, rdo)
		if err == nil {
			status = true
		} else {
			log.Warn("insert fail", err)
			status = false
		}
	} else {
		status = false
	}
	return
}

// 在 pages_articles 更新后， 对比 schedule_revision_info 和 pages_articles，找出过时了的 schedule_revision_info
func (c *MongoDataBase) Get_Overdue_Revision_info() (chan model.ScheduleOfRevisionInfoObj, *sync.WaitGroup) {
	// 忽略重定向的 page
	cur, err := c.collection_pages_articles.Find(c.ctx, bson.M{"redirect": nil, "ns": bson.M{"$nin": []int32{6, 10, 118, 8, 100}}})
	failOnError(err, "查询失败了")

	outchan := make(chan model.ScheduleOfRevisionInfoObj, 10000)
	wg := sync.WaitGroup{}
	wg.Add(1)
	go func() {
		ct := 0
		for cur.Next(c.ctx) {
			ct += 1
			var doc model.PagesArticlesObj
			err := cur.Decode(&doc)
			failOnError(err, "decode 失败了")
			// 判断是否和 schedule_revision_info 中的版本吻合，是否需要更新
			singresult := c.collection_revision_info_schedule.FindOne(c.ctx, bson.M{"_id": doc.ID})
			if singresult.Err() == mongo.ErrNoDocuments {
				outchan <- model.ScheduleOfRevisionInfoObj{
					ID:    doc.ID,
					Title: doc.Title,
				}
			} else {
				var schedule_doc model.ScheduleOfRevisionInfoObj
				err := singresult.Decode(&schedule_doc)
				failOnError(err, "decode 失败了")

				if doc.RevisionID > schedule_doc.LastRevisionID && schedule_doc.FailCount == 0 {
					schedule_doc.Title = doc.Title
					outchan <- schedule_doc
				}
			}
			if ct%100000 == 0 {
				log.Info("current count:", ct)
			}
		}
		close(outchan)
		wg.Done()
	}()

	return outchan, &wg
}

func (c *MongoDataBase) Mark_update_revision_info_fail(ID int64, message string) error {
	opt := options.Update().SetUpsert(true)
	_, err := c.collection_revision_info_schedule.UpdateByID(c.ctx, ID, bson.M{"$inc": bson.M{"failCount": 1}, "$set": bson.M{"message": message}}, opt)
	return err
}

func (c *MongoDataBase) Mark_update_revision_detail_fail(ID primitive.ObjectID, message string) error {
	opt := options.Update().SetUpsert(true)
	_, err := c.collection_revision_detail_schedule.UpdateByID(c.ctx, ID, bson.M{"$inc": bson.M{"failCount": 1}, "$set": bson.M{"message": message}}, opt)
	return err
}

func (c *MongoDataBase) Mark_update_revision_detail_finish(ID primitive.ObjectID) error {
	_, err := c.collection_revision_detail_schedule.UpdateByID(c.ctx, ID, bson.M{"$set": bson.M{"failCount": 0, "finish": true}})
	return err
}

func (c *MongoDataBase) Mark_revision_text_hidden(ID primitive.ObjectID) error {
	c.collection_revision_detail_schedule.UpdateByID(c.ctx, ID, bson.M{"$set": bson.M{"texthidden": true}})
	_, err := c.collection_revision_info.UpdateByID(c.ctx, ID, bson.M{"$set": bson.M{"texthidden": true}})
	return err
}

func (c *MongoDataBase) Bulk_update_revision_info(ID int64, revInfos []*model.RevisionInfoObj) (err error) {
	opts := options.BulkWrite().SetOrdered(false)
	models := []mongo.WriteModel{}
	var schedule_doc model.ScheduleOfRevisionInfoObj
	singresult := c.collection_revision_info_schedule.FindOne(c.ctx, bson.M{"_id": ID})
	if singresult.Err() == mongo.ErrNoDocuments {
		// 没有结果的
		// 找到 LastRevisionID and FirstRevisionID
		var (
			last  = revInfos[0]
			first = revInfos[0]
		)
		for i := range revInfos {
			if revInfos[i].Timestamp.Sub(last.Timestamp) > 0 {
				last = revInfos[i]
			}
			if revInfos[i].Timestamp.Sub(first.Timestamp) < 0 {
				first = revInfos[i]
			}
			models = append(models, mongo.NewUpdateOneModel().SetFilter(bson.M{
				"revid":     revInfos[i].Revid,
				"articleID": revInfos[i].ArticleID,
			}).SetUpdate(bson.M{"$set": *revInfos[i]}).SetUpsert(true))
		}
		retTmp, err := c.collection_revision_info.BulkWrite(c.ctx, models, opts)
		if err != nil {
			log.Println("bulk upsert fail", err)
			log.Printf("%+v", retTmp)
		}
		schedule_doc = model.ScheduleOfRevisionInfoObj{
			ID:                  ID,
			FailCount:           0,
			Update:              time.Now(),
			LastRevisionID:      last.Revid,
			LastRevisionUpdate:  last.Timestamp,
			FirstRevisionID:     first.Revid,
			FirstRevisionUpdate: first.Timestamp,
		}

	} else {
		var schedule_doc model.ScheduleOfRevisionInfoObj
		err = singresult.Decode(&schedule_doc)
		failOnError(err, "decode 失败了")
		// 有结果的

		last := &model.RevisionInfoObj{
			Revid:     schedule_doc.LastRevisionID,
			Timestamp: schedule_doc.LastRevisionUpdate,
		}
		for i := range revInfos {
			if revInfos[i].Timestamp.Sub(last.Timestamp) > 0 {
				last = revInfos[i]
			}
			models = append(models, mongo.NewUpdateOneModel().SetFilter(bson.M{
				"revid":     revInfos[i].Revid,
				"articleID": revInfos[i].ArticleID,
			}).SetUpdate(bson.M{"$set": *revInfos[i]}).SetUpsert(true))
		}
		retTmp, err := c.collection_revision_info.BulkWrite(c.ctx, models, opts)
		if err != nil {
			log.Println("bulk upsert fail", err)
			log.Printf("%+v", retTmp)
		}
		schedule_doc.FailCount = 0
		schedule_doc.Update = time.Now()
		schedule_doc.LastRevisionID = last.Revid
		schedule_doc.LastRevisionUpdate = last.Timestamp
	}
	updateOneOpt := options.Update().SetUpsert(true)
	c.collection_revision_info_schedule.UpdateByID(c.ctx, ID, bson.M{"$set": schedule_doc}, updateOneOpt)
	return
}

// 生成 日期快照
func (c *MongoDataBase) Generate_Snapshot_Article() error {
	timetags := generate_static_time_tag()

	// 遍历所有的完成了的 schedule info doc
	cur, err := c.collection_revision_info_schedule.Find(c.ctx, bson.M{})
	failOnError(err, "查询失败了")

	// 使用管道加速
	doc_chan := make(chan *model.ScheduleOfRevisionInfoObj, 10000000)
	go func() {
		for cur.Next(c.ctx) {
			var docfScheduleRevisionInfo model.ScheduleOfRevisionInfoObj
			cur.Decode(&docfScheduleRevisionInfo)
			if docfScheduleRevisionInfo.FailCount > 0 {
				continue
			}
			doc_chan <- &docfScheduleRevisionInfo
		}
		close(doc_chan)
	}()

	findShowIDTimestampOpt := options.Find().SetProjection(bson.M{"_id": 1, "timestamp": 1, "articleID": 1, "revid": 1})

	// 开启 5 个 处理进程
	thread_count := 5
	wg := sync.WaitGroup{}
	wg.Add(thread_count)
	for i := 0; i <= thread_count; i++ {
		go func() {
			for docfScheduleRevisionInfo := range doc_chan {
				// 找到 snapshot doc
				singresult := c.collection_snapshot_article_revision.FindOne(c.ctx, bson.M{"_id": docfScheduleRevisionInfo.ID})
				var docSnapshot model.SnapshotOfArticleRevisionObj
				if singresult.Err() == mongo.ErrNoDocuments {
					// 没有 snapshot 重新生成
					docSnapshot = model.SnapshotOfArticleRevisionObj{
						ID:       docfScheduleRevisionInfo.ID,
						Birthday: docfScheduleRevisionInfo.FirstRevisionUpdate,
						Snapshot: make(map[string]primitive.ObjectID),
						Update:   time.Now(),
					}
				} else {
					// 有 snapshot
					singresult.Decode(&docSnapshot)
				}
				if docSnapshot.LastSnapshot == time_format(timetags[len(timetags)-1]) {
					// 如果是最新版本，就不用更新了
					continue
				}
				log.Info("start update article ID:", docfScheduleRevisionInfo.ID)

				// 找到这个版本的所有 revision info
				var revision_info_list []*model.RevisionInfoObj
				subCur, _ := c.collection_revision_info.Find(c.ctx, bson.M{"articleID": docfScheduleRevisionInfo.ID, "texthidden": nil}, findShowIDTimestampOpt)
				for subCur.Next(c.ctx) {
					var subDoc model.RevisionInfoObj
					subCur.Decode(&subDoc)
					revision_info_list = append(revision_info_list, &subDoc)
				}
				// 确保按升序排序
				sort.SliceStable(revision_info_list, func(i, j int) bool {
					return revision_info_list[j].Timestamp.After(revision_info_list[i].Timestamp)
				})

				// 更新逻辑
				for _, timeItem := range timetags {
					timgTag := time_format(timeItem)
					if _, state := docSnapshot.Snapshot[timgTag]; !state {
						//  snap 的 reversion info
						var cacheSnapInfo *model.RevisionInfoObj
						// 没有找到 timgTag 的进行处理，找到小于 time 的最大 timTag
						for i := range revision_info_list {
							// 数组中已经没有比 tag 更早的时间了
							if revision_info_list[i].Timestamp.After(timeItem) {
								break
							}
							cacheSnapInfo = revision_info_list[i]
						}
						// 只处理这个时间点的 snap
						if cacheSnapInfo != nil {
							docSnapshot.Snapshot[timgTag] = cacheSnapInfo.ID
							c.Insert_ScheduleOfRevisionDetail(cacheSnapInfo.ID, cacheSnapInfo.Revid, cacheSnapInfo.ArticleID)
						}
					}
				}
				docSnapshot.LastSnapshot = time_format(timetags[len(timetags)-1])
				updateOpt := options.Update().SetUpsert(true)
				_, err = c.collection_snapshot_article_revision.UpdateByID(c.ctx, docSnapshot.ID, bson.M{"$set": docSnapshot}, updateOpt)
			}
			wg.Done()
		}()
	}
	wg.Wait()
	return err
}

// 生成 日期快照
func (c *MongoDataBase) Insert_ScheduleOfRevisionDetail(revisionID primitive.ObjectID, revID, articleID int64) (err error) {

	singresult := c.collection_revision_detail_schedule.FindOne(c.ctx, bson.M{"_id": revisionID})
	// 如果之前没有的就进行更新
	if singresult.Err() == mongo.ErrNoDocuments {
		revisionDetailOneResult := c.collection_revision_detail.FindOne(c.ctx, bson.M{"_id": revisionID})
		var finish bool
		if revisionDetailOneResult.Err() == nil {
			finish = true
		}
		doc := model.ScheduleOfRevisionDetailObj{
			ID:        revisionID,
			Revid:     revID,
			ArticleID: articleID,
			Finish:    finish,
			Update:    time.Now(),
		}
		_, err = c.collection_revision_detail_schedule.InsertOne(c.ctx, doc)
	}
	return
}

// 在 pages_articles 更新后， 对比 schedule_revision_info 和 pages_articles，找出过时了的 schedule_revision_info
func (c *MongoDataBase) Get_Unfinish_Revision_detail() (chan model.ScheduleOfRevisionDetailObj, *sync.WaitGroup) {
	// 忽略重定向的 page
	cur, err := c.collection_revision_detail_schedule.Find(c.ctx, bson.M{"finish": false})
	failOnError(err, "查询失败了")
	outchan := make(chan model.ScheduleOfRevisionDetailObj, 1000)
	wg := sync.WaitGroup{}
	wg.Add(1)
	go func() {
		for cur.Next(c.ctx) {
			var doc model.ScheduleOfRevisionDetailObj
			err := cur.Decode(&doc)
			failOnError(err, "decode 失败了")
			outchan <- doc
		}
		close(outchan)
		wg.Done()
	}()

	return outchan, &wg
}

func (c *MongoDataBase) Get_article_title(ID int64) (title string, err error) {

	singresult := c.collection_pages_articles.FindOne(c.ctx, bson.M{"_id": ID})

	if singresult.Err() == mongo.ErrNoDocuments {
		err = singresult.Err()
	} else {
		var doc model.PagesArticlesObj
		singresult.Decode(&doc)
		title = doc.Title
	}
	return
}
